package com.fanxuankai.boot.canal.listener.autoconfigure;

import com.fanxuankai.boot.canal.listener.MessageListener;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({RocketMQTemplate.class})
@ConditionalOnMissingClass({"com.fanxuankai.boot.mqbroker.producer.EventPublisher"})
@ConditionalOnProperty(name = CanalListenerProperties.ENABLED, matchIfMissing = true)
public class CanalListenerRocketMqAutoConfiguration {
    private static final Logger LOGGER = LoggerFactory.getLogger(CanalListenerRocketMqAutoConfiguration.class);

    @Bean(initMethod = "start")
    public DefaultMQPushConsumer pushConsumer(RocketMQProperties properties, MessageListener messageListener) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(properties.getProducer().getGroup());
        consumer.setNamesrvAddr(properties.getNameServer());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            String topic = consumeConcurrentlyContext.getMessageQueue().getTopic();
            list.forEach(messageExt -> messageListener.onMessage(topic, new String(messageExt.getBody())));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        for (String topic : messageListener.getTopics()) {
            try {
                consumer.subscribe(topic, "*");
            } catch (MQClientException e) {
                LOGGER.error("订阅失败", e);
            }
        }
        return consumer;
    }
}
